package com.xian.java.stream;

import com.mongodb.MongoClient;
import com.mongodb.MongoClientOptions;
import com.mongodb.MongoCredential;
import com.mongodb.ServerAddress;
import com.mongodb.client.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.bson.Document;

/**
 * 获取mongo数据
 */
public class GetMongoMsg {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> ds = env.addSource(new MongoSourceFunction());

        ds.print();


        env.execute();
    }
}


class MongoSourceFunction extends RichSourceFunction<String> {

    private MongoClient mongoClient;
    MongoDatabase db;
    MongoCollection<Document> tableInstant;
    long num = 0; //计算总记录数

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        // 连接到 mongodb 服务
        MongoCredential credential = MongoCredential.createCredential("appuser", "admin", "0IP6k+ERm39lTpjsdBkhohY75WMkyB".toCharArray());   //验证对象
        MongoClientOptions options = MongoClientOptions.builder().build();     //连接操作对象
        mongoClient = new MongoClient(new ServerAddress("172.18.5.100",20017),credential,options);   //连接对象


        System.out.println(mongoClient);
        //验证库
        db = mongoClient.getDatabase("carorder");

        //获取某个表
        tableInstant = db.getCollection("OrderExpCarInstant");







    }

    @Override
    public void run(SourceContext<String> ctx) throws Exception {

        FindIterable<Document> documents = tableInstant.find();
        MongoCursor<Document> it = documents.iterator();


        while (it.hasNext()) {
            num++; //记录数+1
            Document doc = it.next();
            //System.out.println(doc);
            //不打印 发送出去
            ctx.collect(doc.toJson());
        }
    }

    @Override
    public void cancel() {
        System.out.println("取消。。。");
    }

    @Override
    public void close() throws Exception {
        super.close();
        System.out.println("关闭资源。。。");
        System.out.println("总记录数为："+num);
        System.out.println("官方计数为："+tableInstant.countDocuments());
        mongoClient.close();
    }
}